package com.gitee.huanminabc.jcommon.multithreading.delay;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.Objects;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

@Slf4j
@Data
public class ConsumerDelayed<T> implements Runnable {
    // 延时队列 ,消费者从其中获取消息进行消费
    private final DelayQueue<MessageDelayed<T>> queue;
    private final Consumer<MessageDelayed<T>> consumer;
    private volatile String threadName;
    private final AtomicBoolean running = new AtomicBoolean(true);
    private final int maxRetries;
    private final long retryDelayMs;

    public ConsumerDelayed(DelayQueue<MessageDelayed<T>> queue, Consumer<MessageDelayed<T>> consumer) {
        this(queue, consumer, 3, 1000);
    }

    public ConsumerDelayed(DelayQueue<MessageDelayed<T>> queue, Consumer<MessageDelayed<T>> consumer, 
                          int maxRetries, long retryDelayMs) {
        Objects.requireNonNull(queue, "Queue cannot be null");
        Objects.requireNonNull(consumer, "Consumer cannot be null");
        this.queue = queue;
        this.consumer = consumer;
        this.maxRetries = maxRetries;
        this.retryDelayMs = retryDelayMs;
    }

    @Override
    public void run() {
        log.info("Consumer thread started: {}", Thread.currentThread().getName());
        
        while (running.get()) {
            try {
                // 取出到期的元素,并且从队列中删除
                MessageDelayed<T> message = queue.take();
                processMessageWithRetry(message);
            } catch (InterruptedException e) {
                log.info("Consumer thread interrupted: {}", Thread.currentThread().getName());
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("Unexpected error in consumer thread: {}", e.getMessage(), e);
                // 短暂休眠避免CPU空转
                try {
                    Thread.sleep(100);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
        
        log.info("Consumer thread stopped: {}", Thread.currentThread().getName());
    }

    private void processMessageWithRetry(MessageDelayed<T> message) {
        int retryCount = 0;
        while (retryCount < maxRetries && running.get()) {
            try {
                consumer.accept(message);
                return; // 成功处理，退出重试
            } catch (Exception e) {
                retryCount++;
                log.warn("Failed to process message (attempt {}/{}): {}", 
                        retryCount, maxRetries, e.getMessage());
                
                if (retryCount < maxRetries) {
                    try {
                        Thread.sleep(retryDelayMs);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }
        
        if (retryCount >= maxRetries) {
            log.error("Failed to process message after {} retries: {}", maxRetries, message);
        }
    }

    /**
     * 优雅关闭消费者
     */
    public void shutdown() {
        running.set(false);
        log.info("Shutdown signal sent to consumer: {}", threadName);
    }

    /**
     * 检查消费者是否正在运行
     */
    public boolean isRunning() {
        return running.get();
    }

    /**
     * 启动消费者线程
     */
    public void start() {
        Thread thread;
        if (threadName == null) {
            thread = new Thread(this);
        } else {
            thread = new Thread(this, threadName);
        }
        thread.setDaemon(true);
        thread.start();
        log.info("Consumer thread started: {}", thread.getName());
    }
}
